package org.nbict.iot.trash;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;
import org.nbict.iot.persist.pool.DbRepository;
import org.nbict.iot.persist.repo.MongoRepo;

import java.util.HashMap;
import java.util.Map;

/**
 * S0-2
 * 报文体内容分发:
 *  01-整车
 *  02-驱动电机
 *  03-燃料电池
 *  04-发动机
 *  05-车辆位置
 *  06-极值数据
 *  07-报警数据
 *  08-可充电储能装置电压
 *  09-可充电储能装置温度
 * Created by songseven on 18/6/13.
 */
@Deprecated
public class S02Bolt extends BaseBasicBolt {

    private MongoRepo repo;

    public void prepare(Map stormConf, TopologyContext context) {
        repo = (MongoRepo) DbRepository.MONGO.getInstance();
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

        JSONObject o = (JSONObject) input.getValueByField("parsed");
        JSONObject header = o.getJSONObject("car_packet_header");
        JSONObject body = o.getJSONObject("car_packet_body");

        if (body == null) return;

        JSONObject upload = body.getJSONObject("upload");
        if (upload == null) return; //只记录实时数据及补发数据

        String vin = header.getString("car_vin");
        Long timestamp = body.getLong("timestamp");
        String companyId = o.getString("company_id");

        //更新地图1
        Map query = new HashMap();
        query.put("car_vin", vin);
        BasicDBObject vehicle = repo.get("vehicle_map", query);
        if (vehicle == null) return; //非注册车辆，不更新

        boolean isUpdate = false;
        if (vehicle.getLong("timestamp") < timestamp) {
            vehicle.put("timestamp", timestamp);
            isUpdate = true;
        }

        if (!isUpdate) return; //不需更新

        //更新车辆实时信息
        for (String tag : upload.keySet()) {
            Map info = JSONObject.toJavaObject((JSON) upload.get(tag), Map
                    .class);
            Map stored = new HashMap();
            stored.put("car_vin", vin);
            stored.put("timestamp", timestamp);
            stored.put("platform_id", companyId);
            stored.put("tag", tag);
            stored.put("info", info);

            Map query1 = new HashMap();
            query1.put("car_vin", vin);
            query1.put("tag", tag);

            BasicDBObject object = new BasicDBObject(stored);
            repo.upsert("car", query1, object);

            //TODO 需要增加更新经纬度及状态
            if (tag.equals("01")) { //整车
                vehicle.put("car_status", info.get("status")); //车辆状态
                vehicle.put("charge_status", info.get("charged_state")); //充电状态
            }
            if (tag.equals("05")) {
                vehicle.put("lon", info.get("lon"));
                vehicle.put("lat", info.get("lat"));
            }
            if (tag.equals("07")) { //告警
                vehicle.put("alarm_status", info.get("alarm_level"));
            }
        }

        //更新地图2
        repo.upsert("vehicle_map", query, vehicle);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        //declarer.declare(new Fields("payload"));
    }
}
